package com.lvmama.rhino.analyze.client

import com.google.common.base.CharMatcher
import com.lvmama.rhino.analyze.processing.{H5PageMonitorChannel, H5PageMonitorNetworkType, H5PageMonitorOperators, H5PageMonitorRegion}
import com.lvmama.rhino.common.entity.JobConfig
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.spark.sql.functions._
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions.ImplicitInsert
import com.lvmama.rhino.common.utils.logging.LazyLogging
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.{Column, SQLContext}

/**
  * Created by Administrator on 2016/12/7.
  */
class H5PageMonitor(config: JobConfig, path: String) extends SparkApplication with LazyLogging{
  override var appName: String = "H5PageMonitorJob"
  override var sparkConfig: Map[String, String] = config.spark

  def processUrl = udf((url: String) => {
    var ext_url = CharMatcher.JAVA_DIGIT.replaceFrom(url, "*")
    ext_url = ext_url.replaceAll("[*]+", "*")
    ext_url
  })

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)
    sparkConfig += ("spark.master" -> "local[3]")
    withSparkContext { sc =>
      val sqlCtx = new SQLContext(sc)
      import sqlCtx.implicits._
      val log = sc.textFile(path).map(l => Utils.initPageMonitor(l)).toDF()
      val result = log.filter(!col("createDate").equalTo(""))
      val monitorData=result.select(col("province"), col("city"), col("operators"), col("categoryId").as("category_id"),
      col("networkType").as("network_type"), col("loadTime").as("load_time"),col("url").as("url"),
      col("islv").as("is_lv"), col("createDate").as("create_date"))

      monitorData.insertDF2MysqlToWolverine("h5_page_monitor")

      val h5MonitorData = monitorData.withColumn("time_interval",when(col("load_time").between(0,1),0).when(col("load_time").between(1,2),1)
        .when(col("load_time").between(2,3),2).when(col("load_time").between(3,4),3).when(col("load_time").between(4,5),4)
        .when(col("load_time").between(5,8),5).when(col("load_time").>(8),6))
        .withColumn("oper_date",col("create_date").substr(0,10))
        .withColumn("ext_url",processUrl(col("url").cast(StringType)))

      //h5首屏时间按渠道区分
      logger.info(s"H5PageMonitor begin process H5PageMonitorChannel")
      H5PageMonitorChannel.process(sc, h5MonitorData)

      //h5首屏时间按运营商区分
      logger.info(s"H5PageMonitor begin process H5PageMonitorOperators")
      H5PageMonitorOperators.process(sc, h5MonitorData)

      //h5首屏时间按网络制式区分
      logger.info(s"H5PageMonitor begin process H5PageMonitorNetworkType")
      H5PageMonitorNetworkType.process(sc, h5MonitorData)

      //h5首屏时间按地区区分
      logger.info(s"H5PageMonitor begin process H5PageMonitorRegion")
      H5PageMonitorRegion.process(sc, h5MonitorData)
    }
  }
}
  object H5PageMonitor{
    def apply(path: String): Unit = {
      val config = JobConfig()
      new H5PageMonitor(config, path).execute()
    }
  }